package com.shujia.spark.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo1WordCountOnSQL {
  def main(args: Array[String]): Unit = {
    /**
     * 1 创建spark sql 环境
     */
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("wc")
      .config("spark.sql.shuffle.partitions", 1) //指定spark sql shuffle之后rdd的分区数
      .getOrCreate()

    //通过SparkSession获取SparkContext
    //val sc: SparkContext = spark.sparkContext


    /**
     * 1 读取数据构建DataFrame
     * DataFrame: 底层是RDD,DataFrame 在RDD的基础上增加了表结构,可以写SQL和DSl
     */

    val linesDF: DataFrame = spark
      .read
      .format("csv") // 读取数据的格式
      .option("sep", "|") // 字段的分隔符
      .schema("line STRING") // 指定字段名和字段类型
      .load("data/words.txt") //指定读取数据的路径

    //打印表结构
    linesDF.printSchema()
    linesDF.show()

    /**
     * 1 通过sql统计单词的数量
     *
     * 在spark sql中,会将sql语句转换成RDD的代码去执行
     *
     */
    //将DF转换成临时视图
    linesDF.createOrReplaceTempView("lines")

    //写sql统计单词的数量
    val resultDF: DataFrame = spark.sql(
      """
        |select
        |word,
        |count(1) as num
        |from (
        |       select explode(split(line,',')) as word
        |       from lines
        |) as a
        |group by word
        |""".stripMargin)

    //保存结果
    resultDF
      .write
      .format("csv") //指定保存数据的格式
      .option("sep", ",") //指定分割负
      .mode(SaveMode.Overwrite) //指定覆盖写入
      .save("data/word_count") //指定保存数据的路径

  }
}
